package com.smsc.headend.task.engine.service.push.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.smsc.headend.common.enums.DeviceType;
import com.smsc.headend.common.utils.*;
import com.smsc.headend.module.asset.dmo.ComConfigDMO;
import com.smsc.headend.module.asset.dmo.MeterConfigDMO;
import com.smsc.headend.module.asset.entity.Dcu;
import com.smsc.headend.module.asset.entity.DeviceRelation;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.asset.entity.UnRegisterDevice;
import com.smsc.headend.module.data.dmo.GroupSubDataItem;
import com.smsc.headend.module.data.enums.OperationGroup;
import com.smsc.headend.module.data.enums.OperationName;
import com.smsc.headend.module.event.dto.DeviceNotificationDto;
import com.smsc.headend.module.event.entity.*;
import com.smsc.headend.module.protocol.cosem.bo.CosemData;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.report.entity.LogUnRegisterMeterHis;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.*;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.module.task.proto.TaskEventDataProto;
import com.smsc.headend.module.template.alarmfilter.entity.AlarmObjEntity;
import com.smsc.headend.module.template.alarmfilter.entity.AlarmTypeEntity;
import com.smsc.headend.task.engine.facade.DataItemFacade;
import com.smsc.headend.task.engine.mapper.*;
import com.smsc.headend.task.engine.service.CosemXmlService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.push.DcuRegisterTaskService;
import com.smsc.headend.task.engine.service.push.EventService;
import com.smsc.headend.task.engine.service.task.TaskPacketService;
import com.smsc.headend.task.engine.service.task.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.assertj.core.util.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.sql.DataSource;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * @author liangli
 * @date 2020/11/2
 */
@Service
@Slf4j
public class EventServiceImpl implements EventService {
    @Autowired
    FeignAssetManagementService assetManagementService;
    @Autowired
    CosemXmlService cosemXmlService;
    @Autowired
    RdEventMeterMapper rdEventMeterMapper;
    @Autowired
    RdEventMeterDataMapper rdEventMeterDataMapper;
    @Autowired
    RdEventDcuMapper rdEventDcuMapper;
    @Autowired
    RdEventDcuDataMapper rdEventDcuDataMapper;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    AssetService assetService;
    @Autowired
    DataSource dataSource;
    @Autowired
    AlarmObjMapper alarmObjMapper;
    @Autowired
    AlarmTypeMapper alarmTypeMapper;
    @Autowired
    TaskPacketService packetService;
    @Autowired
    DataItemFacade dataItemFacade;
    @Autowired
    TaskService taskService;
    @Autowired
    RdAlarmMeterMapper rdAlarmMeterMapper;

    @Autowired
    DeviceRelationMapper deviceRelationMapper;

    @Autowired
    DcuRegisterTaskService registerTaskService;

    Set<Long> wsDcuRegisterDataItemIds = Collections.unmodifiableSet(new HashSet<Long>() {{
        add(1100400001L); // Push Setup –Joining Notification
        add(1100400002L); // Push Setup – Registered Notification
        add(1100400003L); // Push Setup – Re-registered Notification
    }});

    Set<Integer> needAsciiDecodeEventTypes = Collections.unmodifiableSet(new HashSet<Integer>() {{
        add(110001103);//Meter Clock Diff Over Limit
        add(110001106);//Meter Read Fail
    }});

    Set<Long> wsPushAlarmDataItemsIds = Collections.unmodifiableSet(new HashSet<Long>() {{
        add(1020400004L); // Push setup - On Alarm
    }});

    Map<Long, Long> adMapAr = MapUtil.builder(1020010017102L, 1020010015102L).put(1020010021102L, 1020010019102L).map();

    @Override
    public Integer getTaskEventType(Task task, Integer eventCode, Long lbDataItemId, Long deviceTypeId) {
        Long modelId;
        if (task.getMeterId() == null) {
            ComConfigDMO comConfigDMO = assetService.getComConfigById(task.getComId());
            modelId = comConfigDMO.getModelId();
        } else {
            Meter meter = assetService.getMeterById(task.getMeterId());
            modelId = meter.getModelId();
        }
        return assetManagementService.getModelEventType(modelId, eventCode, lbDataItemId, deviceTypeId);
    }

    @Override
    public String getEventName(Long eventTypeId) {
        return assetManagementService.getEventName(eventTypeId);
    }

    /**
     * Cosem协议 事件解析
     *
     * @param cosemData
     * @param modelId
     * @param deviceType
     * @param deviceId
     * @param dataItemId
     */
    @Override
    public void analyseCOSEMEvent(CosemData cosemData, Long modelId, Long deviceType, Long deviceId, Long dataItemId) {
        Date uploadDate;
        List<CosemData> notifications = CollUtil.newArrayList();
        if (dataItemId == null) {
            CosemData dateData = cosemData.getChildAt(1);
            uploadDate = (Date) cosemXmlService.getData(dateData.getType(), dateData.getValue());
            notifications.add(cosemData.getChildAt(2).getChildAt(0));
        } else {
            uploadDate = new Date();
            notifications = cosemData.getChild();
        }
        if(null != dataItemId && wsDcuRegisterDataItemIds.contains(dataItemId)) {
            handleWasionDcuRegisterNotify(dataItemId, notifications, deviceId);
        } else {
            if (null != dataItemId && wsPushAlarmDataItemsIds.contains(dataItemId)) {
                notifications = CollUtil.newArrayList();
                CosemData dateData = cosemData.getChildAt(1);
                uploadDate = (Date) cosemXmlService.getData(dateData.getType(), dateData.getValue());
                notifications.add(cosemData.getChildAt(2).getChildAt(0));
            }

            DeviceNotificationDto notificationItemList = assetManagementService.getDeviceNotificationItem(modelId, dataItemId);
            if (notificationItemList == null) {
                log.error("Device Event: no notification obj, deviceType={},deviceId={},modelId={}", deviceType, deviceId, modelId);
                return;
            }
            Date finalUploadDate = uploadDate;
            notifications.stream().forEach(notification -> processNotification(modelId, deviceType, deviceId, dataItemId, finalUploadDate, notification, notificationItemList));
        }
    }

    @Override
    public void analyseBPDBCOSEMEvent(CosemData cosemData, Long modelId, Long deviceType, Long deviceId, Long dataItemId) {
        Date uploadDate;
        DeviceNotificationDto notificationItemList = null;
        try {
            notificationItemList = assetManagementService.getDeviceNotificationItem(modelId, dataItemId);
        } catch (Exception e) {
            log.error("Exception {}", e);
        }
        List<CosemData> childList = cosemData.getChild();
        List<CosemData> notifications = CollUtil.newArrayList();
        for (CosemData childData:childList) {
            CosemData dateData = childData.getChild().get(0);
            CosemData eventCodeData = childData.getChild().get(1);
            String timeString = (String) cosemXmlService.getData(dateData.getType(), dateData.getValue());
            Integer eventCode = Convert.toInt(cosemXmlService.getData(eventCodeData.getType(), eventCodeData.getValue()));
            uploadDate = CosemUtils.parseOctStringDataTime(timeString);

            notifications.add(childData);
            CosemData cData = new CosemData();
            cData.setChild(notifications);
            processBPDBNotification(modelId, deviceType, deviceId, dataItemId, uploadDate, eventCode, cData, notificationItemList);
        }

    }

    private void handleWasionDcuRegisterNotify(Long dataItemId, List<CosemData> notifications, Long deviceId) {
        switch (Convert.toInt(dataItemId)) {
            case 1100400001: // Push Setup –Joining Notification
            case 1100400003: // Push Setup – Re-registered Notification
                // Nesco项目不需要暂不实现
                log.error("no implementation currently");
                break;
            case 1100400002: // Push Setup – Registered Notification
                handleRegisterNotify(notifications, deviceId);
                break;
        }
    }

    private void handleRegisterNotify(List<CosemData> notifications, Long deviceId) {
        List<CosemData> cosemChildDataList = notifications.get(2) == null ?
                null : notifications.get(2).getChildAt(0) == null ?
                null : notifications.get(2).getChildAt(0).getChildAt(0) == null ?
                null : notifications.get(2).getChildAt(0).getChildAt(0).getChildAt(1) == null ?
                null : notifications.get(2).getChildAt(0).getChildAt(0).getChildAt(1).getChild();
        if (CollUtil.isNotEmpty(cosemChildDataList)) {
            for (CosemData cosemData: cosemChildDataList) {
                // 该meterId为表计资产号
                String meterId = Convert.toStr(cosemXmlService.getData(cosemData.getChildAt(0).getType(), cosemData.getChildAt(0).getValue()));
                String eui64 = Convert.toStr(cosemXmlService.getData(cosemData.getChildAt(1).getType(), cosemData.getChildAt(1).getValue()));
                Integer status = Convert.toInt(cosemXmlService.getData(cosemData.getChildAt(2).getType(), cosemData.getChildAt(2).getValue()));
                Date date = (Date)(cosemXmlService.getData(cosemData.getChildAt(3).getType(), cosemData.getChildAt(3).getValue()));
                log.info("the register notification info: meterId {}, eui64 {} status {}, date {}", meterId, eui64, status, date.getTime() / 1000);

                // Status 1: New meter; 2: Registered and online meter; 3: Registered and offline meter; 4: Rejected meter (只需处理New Meter的)
                if (status == 1) {
                    UnRegisterDevice unRegisterDevice = assetManagementService.getUnRegisterRecordByDeviceSn(meterId);
                    String autoRegisterSwitch = assetManagementService.getPropertyValue("AUTO_REGISTER_SWITCH");
                    if (!Convert.toBool(autoRegisterSwitch, false)) {
                        log.info("the GUI auto register switch is disabled, no device task will generate");
                    } else {
                        if (null == unRegisterDevice) {
                            //record 没记录马上生成任务
                            handleRegisterTask(meterId, deviceId);
                        } else if(SystemClock.now() / 1000 - unRegisterDevice.getTvUpdate() > 300){
                            // 大于5分钟产生task 防止短时间内dcu重复上报生成task任务
                            handleRegisterTask(meterId, deviceId);
                        }
                    }
                    // 任务不产生也要插入未注册的设备记录
                    assetManagementService.saveUnRegisterRecord(meterId, deviceId, SystemClock.now() / 1000, "Push Notify");
                } else {
                    log.info("The status of the meter {} is not new meter, no action need to do", meterId);
                }
            }
        } else {
            log.error("incorrect register notification {}", notifications);
        }
    }

    private void handleRegisterTask(String meterId, Long deviceId) {
        Meter meter = assetManagementService.getMeterByMfgSn(meterId);
        if (null == meter) {
            log.info("the meter not exist in udis {}", meterId);
        } else {
            DeviceRelation deviceRelation = deviceRelationMapper.selectOne(Wrappers.<DeviceRelation>lambdaQuery()
                    .eq(DeviceRelation::getDeviceId, meter.getMeterId()).eq(DeviceRelation::getDeviceType, DeviceType.METER.getId()));
            if (null != deviceRelation) {
                // 1. 电表在该DCU下有挂接关系 如果已经是mounted状态不做任何操作,否则需要下发白名单到dcu上面
                // 2. 电表在该DCU下无挂接关系 认为是漂移到其他DCU上 或者是未挂接到任何DCU上.如果是未挂接到任何集中器上,则下发白名单. 如果是漂移到其他集中器，则需删除之前集中器上的白名单,解除meter relation关系。并下发新的白名单到当前DCU上面并建立新的meter relation关系
                if (deviceRelation.getParentId().equals(deviceId)) {
                    if (meter.getMountStatus() != null && meter.getMountStatus() == Meter.MeterMountStatus.Mounted.getCode()) {
                        log.info("the meter {} mounted to the dcu {}, no action need to do", meterId, deviceId);
                    } else {
                        insertSetWhiteListTask(deviceRelation.getParentId(), meter);
                        log.info("the meter {} not mounted to the dcu {}, insert setWhiteList task", meterId, deviceId);
                    }
                } else {
                    // 2-1 找出该meter是否跟旧的dcu有关系,有的话旧的dcu删除白名单
                    insertDeleteWhiteListTask(deviceRelation.getParentId(), meter);

                    // 2-2 下发白名单到新的dcu上
                    insertSetWhiteListTask(deviceId, meter);
                }
            } else {
                // 3: 表计跟dcu无任何关系
                // 3-1: 下发白名单到当前dcu上面, 成功后在handler里面挂接dcu与meter之间的关系
                log.error("the meter have no meter relation {}", meterId);
                insertSetWhiteListTask(deviceId, meter);
            }
        }
    }

    private void insertSetWhiteListTask(Long deviceId, Meter meter) {
        Dcu dcu = assetManagementService.getDcuById(deviceId);
        if (dcu != null) {
            Task task = registerTaskService.dcuSetMeterListTask(meter, dcu.getDcuId());
            log.info("insert register setMeterList task {}", JSONUtil.toJsonStr(task));
            assetManagementService.saveUnregisterMeterHis(meter.getMeterId(), dcu.getDcuId(), task.getTaskNo(), LogUnRegisterMeterHis.ActionType.SetMeterList.getCode(),
                    LogUnRegisterMeterHis.UnRegisterSrc.Auto.getCode(), LogUnRegisterMeterHis.UnRegisterStatus.PROCESSING.getCode(), SystemClock.now() / 1000, SystemClock.now() / 1000, "");
            taskService.sendTaskToGenerator(task);
        } else {
            log.info("the current dcu not exist {}", deviceId);
        }
    }

    private void insertDeleteWhiteListTask(Long deviceId, Meter meter) {
        Dcu oldDcu = assetManagementService.getDcuById(deviceId);
        if (oldDcu != null) {
            Task task = registerTaskService.dcuDeleteMeterListTask(meter, oldDcu.getDcuId());
            log.info("insert register deleteMeterList task {}", JSONUtil.toJsonStr(task));
            assetManagementService.saveUnregisterMeterHis(meter.getMeterId(), oldDcu.getDcuId(), task.getTaskNo(), LogUnRegisterMeterHis.ActionType.DeleteMeterList.getCode(),
                    LogUnRegisterMeterHis.UnRegisterSrc.Auto.getCode(), LogUnRegisterMeterHis.UnRegisterStatus.PROCESSING.getCode(), SystemClock.now() / 1000, SystemClock.now() / 1000, "");
            taskService.sendTaskToGenerator(task);
        } else {
            log.info("the old dcu not exist {}", deviceId);
        }
    }

    @Override
    public void processBPDBNotification(Long modelId, Long deviceType, Long deviceId, Long dataItemId, Date uploadDate, Integer eventCode, CosemData notificationDataValue, DeviceNotificationDto notificationItemList) {
        Integer eventType = assetManagementService.getModelEventType(modelId, eventCode, dataItemId, deviceType);
        if (eventType == null) {
            log.error("fail to get model {} eventType by {}", modelId, eventCode);
            return;
        }
        if (notificationItemList != null) {
            // 存子数据解析
            dataItemId = null;
        } else {
            log.debug("Device Event: no notification obj, deviceType={},deviceId={},modelId={}", deviceType, deviceId, modelId);
        }
        savePushUpEventData(eventCode, eventType, notificationDataValue, uploadDate, notificationItemList, deviceId, deviceType, dataItemId);
    }

    private void processNotification(Long modelId, Long deviceType, Long deviceId, Long dataItemId, Date uploadDate, CosemData notificationDataValue, DeviceNotificationDto notificationItemList) {
        Integer eventCode = getEventCode(notificationItemList, notificationDataValue, dataItemId);
        //推送报文中包含EventCode和数据项
        if (eventCode != null) {
//            log.info("get Event code {}", eventCode);
            Integer eventType = assetManagementService.getModelEventType(modelId, eventCode, dataItemId, deviceType);
            if (eventType == null) {
                log.error("fail to get model {} eventType by {}", modelId, eventCode);
                return;
            }
            savePushUpEventData(eventCode, eventType, notificationDataValue, uploadDate, notificationItemList, deviceId, deviceType, dataItemId);
        } else {
            // Alarm Describer
            Map<String, Long[]> lbDataItems = MapUtil.newHashMap();
            for (GroupSubDataItem groupSubDataItem : notificationItemList.getObjectList()) {
                CosemData data = notificationDataValue.getChildAt(0).getChildAt(Convert.toInt(groupSubDataItem.getIndex()) - 1);
                Object value = cosemXmlService.getData(data.getType(), data.getValue());
                String str = Long.toHexString(Convert.toLong(value)); // 获取的Uint32转成了String
                str = CustomStringUtils.getNumberStringWithCount(str, 8); // 转成8位二进制
                str = CustomStringUtils.hexStringToBitString(str); //转成32位0或者1的二进制;
                if (!str.contains("1")) {
                    continue;
                }
                //处理Alarm Descriptor
                String taskNo = clearAdAndAr(deviceId, groupSubDataItem.getSubDataItemId(), str);
                str = StrUtil.reverse(str);
                AlarmObjEntity alarmObj = alarmObjMapper.selectOne(Wrappers.<AlarmObjEntity>lambdaQuery().eq(AlarmObjEntity::getSubDataItemId, groupSubDataItem.getSubDataItemId()));
                List<AlarmTypeEntity> alarmTypes = alarmTypeMapper.selectList(Wrappers.<AlarmTypeEntity>lambdaQuery().eq(AlarmTypeEntity::getAlarmTypeId, alarmObj.getAlarmObjId()).orderByAsc(AlarmTypeEntity::getAlarmBit));
                if (CollUtil.isEmpty(alarmTypes)) {
                    log.error("Device Event: no alarmTypes, deviceType={},deviceId={},modelId={}", deviceType, deviceId, modelId);
                    continue;
                }
                List<String> matchedAlarms = CollUtil.newArrayList();
                for (int i = 0; i < str.toCharArray().length; i++) {
                    if (str.toCharArray()[i] == '1') {
                        AlarmTypeEntity alarmType = alarmTypes.get(i);
                        if (alarmType == null) {
                            continue;
                        }
                        matchedAlarms.add(alarmType.getAlarmTypeDesc());
                        // alarm 入库
                        String alarmLogSeq = saveRdAlarmMeter(deviceId, alarmType.getAlarmTypeId(), alarmType.getAlarmBit(), uploadDate, taskNo);
                        if(StrUtil.isNotEmpty(alarmType.getLbDataItemIds())) {
                            String[] ids = alarmType.getLbDataItemIds().split(",");
                            lbDataItems.put(alarmLogSeq, Arrays.stream(ids).map(id -> Convert.toLong(id)).toArray(size -> new Long[size]));
                        }
                    }
                }
                if (CollUtil.isNotEmpty(matchedAlarms)) {
                    Meter meter = assetService.getMeterById(deviceId);
                    Map<String, Object> map = MapUtil.newHashMap();
                    map.put("customerSn", meter.getCustomerSn());
                    map.put("tv", uploadDate.getTime());
                    map.put("alarms", CollUtil.join(matchedAlarms, ","));
                    kafkaTemplate.send(TaskKafkaTopic.ALARM_TO_ADAPTER_FOR_EMAIL, JSONUtil.toJsonStr(map).getBytes(StandardCharsets.UTF_8));
                }
            }
            if (CollUtil.isNotEmpty(lbDataItems)) {
                PullEventData data = new PullEventData();
                data.setDeviceId(deviceId);
                data.setDeviceType(deviceType);
                data.setLbDataItems(lbDataItems);
                kafkaTemplate.send(TaskKafkaTopic.PULL_EVENT_TASK, JSONUtil.toJsonStr(data).getBytes(StandardCharsets.UTF_8));
            }
        }
    }

    private String clearAdAndAr(Long meterId, Long subDataItemId, String value) {
        String alarmFilterValue = CustomStringUtils.bitStringToHexString(value); //二进制转成十六进制
        Long alarmLongValue = Long.parseLong(alarmFilterValue, 16);
        MeterConfigDMO meterConfigDMO = assetManagementService.getMeterConfigById(meterId);

        Task task = new Task();
        task.setComId(meterConfigDMO.getComId());
        task.setMeterId(meterId);
        task.setTaskType(TaskType.AlarmDescriptor.toString());
        task.setIsODR(true);
        task.setMessageId(UUID.randomUUID().toString());
        task.setPriority(Task.Priority.NORMAL.getPriorityCode());
        String taskNo = UUID.randomUUID().toString();
        task.setTaskNo(taskNo);

        List<SubTaskGroup> subTaskGroups = CollUtil.newArrayList();
        SubTaskGroup subTaskGroup = new SubTaskGroup();
        subTaskGroup.setGroupId(OperationGroup.GeneralParametersSet.getId().longValue());
        List<AtomicTask> atomicTasks = new ArrayList<>();

        AtomicTask adAtomicTask = new AtomicTask();
        adAtomicTask.setSeq(0);
        adAtomicTask.setGroupId(subTaskGroup.getGroupId());
        adAtomicTask.setAtomicTaskNo(UUID.randomUUID().toString());
        adAtomicTask.setTaskNo(taskNo);
        adAtomicTask.setOperationId(OperationName.GeneralParamSet.getId().longValue());
        Map<String, Object> atomicTaskParam = MapUtil.newHashMap();
        atomicTaskParam.put("subDataItemId", subDataItemId);
        atomicTaskParam.put("value", alarmLongValue);
        adAtomicTask.setAtomicTaskParam(JSONUtil.toJsonStr(atomicTaskParam));
        atomicTasks.add(adAtomicTask);

        // 先不清除AR
        /*AtomicTask arAtomicTask = new AtomicTask();
        arAtomicTask.setSeq(1);
        arAtomicTask.setGroupId(subTaskGroup.getGroupId());
        arAtomicTask.setAtomicTaskNo(UUID.randomUUID().toString());
        arAtomicTask.setTaskNo(taskNo);
        arAtomicTask.setOperationId(OperationName.GeneralParamSet.getId().longValue());
        Map<String, Object> atomicTaskParam1 = MapUtil.newHashMap();
        atomicTaskParam1.put("subDataItemId", adMapAr.get(subDataItemId));
        atomicTaskParam1.put("value", alarmLongValue);
        arAtomicTask.setAtomicTaskParam(JSONUtil.toJsonStr(atomicTaskParam1));
        atomicTasks.add(arAtomicTask);*/

        subTaskGroup.setSubTasks(atomicTasks);
        subTaskGroups.add(subTaskGroup);
        task.setSubTaskGroups(subTaskGroups);
        log.info("clear ad and ar task: {}", JSONUtil.toJsonStr(task));
        taskService.sendTaskToGenerator(task);
        return taskNo;
    }

    private Integer getEventCode(DeviceNotificationDto notificationItemList, CosemData notificationDataValue, Long dataItemId) {
        if (notificationItemList == null) {
            return null;
        }
        Integer i = 0;
        Boolean hasEventCode = false;
        for (GroupSubDataItem groupSubDataItem : notificationItemList.getObjectList()) {
            if (groupSubDataItem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.EVENT_CODE.getCode())) {
                hasEventCode = true;
                break;
            }
            i++;
        }
        if (!hasEventCode) {
            return null;
        }
        CosemData cosemData;
        if (dataItemId == null) {
            cosemData = notificationDataValue.getChildAt(0).getChildAt(i);
        } else {
            cosemData = notificationDataValue.getChildAt(i);
        }
        return Convert.toInt(cosemXmlService.getData(cosemData.getType(), cosemData.getValue()));
    }

    private TaskEventData savePushUpEventData(Integer eventCode, Integer type, CosemData notificationDataValue, Date uploadDate, DeviceNotificationDto notificationItemList, Long deviceId, Long deviceType, Long dataItemId) {
        if (deviceType == DeviceType.METER.getId().longValue() || deviceType == DeviceType.GPRS.getId().longValue()) {
            RdEventMeter rdEventMeter = new RdEventMeter();
            rdEventMeter.setLogSeq(UUID.randomUUID().toString());
            rdEventMeter.setEventType(type);
            rdEventMeter.setMeterId(deviceId);
            rdEventMeter.setEventCode(eventCode);
            rdEventMeter.setTv(getEventTv(uploadDate, notificationDataValue, notificationItemList, dataItemId));
            rdEventMeter.setTvUpdate(System.currentTimeMillis() / 1000);
            //推送固定使用data_item_id
            rdEventMeter.setLbDataItemId(1024000001l);
            rdEventMeter.setReceiveType(1);
            // rdEventMeterMapper.insert(rdEventMeter);
            if (dataItemId == null) {
                List<RdEventMeterData> subData = saveEventSubData(rdEventMeter.getLogSeq(), notificationDataValue, notificationItemList);
                rdEventMeter.setSubData(subData);
            }
            meterEventSendToDDS(rdEventMeter);
        } else if (deviceType == DeviceType.DCU.getId().longValue()) {
            RdEventDcu rdEventDcu = new RdEventDcu();
            rdEventDcu.setLogSeq(UUID.randomUUID().toString());
            rdEventDcu.setEventType(type.longValue());
            rdEventDcu.setEventCode(eventCode);
            rdEventDcu.setDcuId(deviceId);
            if (dataItemId == null) {
                rdEventDcu.setTv(uploadDate.getTime() / 1000);
            } else {
                rdEventDcu.setTv(getEventTv(uploadDate, notificationDataValue, notificationItemList, dataItemId));
            }
            rdEventDcu.setTvUpdate(System.currentTimeMillis() / 1000);
            rdEventDcu.setReceiveType(1);
            // rdEventDcuMapper.insert(rdEventDcu);
            if (dataItemId == null) {
                List<RdEventDcuData> subData = saveDcuEventSubData(rdEventDcu.getLogSeq(), type,  notificationDataValue, notificationItemList);
                rdEventDcu.setSubData(subData);
            }
            meterEventSendToDDS(rdEventDcu);
        } else {
            log.error("deviceType not support {}", deviceType);
        }
        return null;
    }

    private void meterEventSendToDDS(AbstractRdEvent rdEvent) {
        TaskEventData taskEventData = new TaskEventData();
        if (rdEvent instanceof RdEventMeter) {
            RdEventMeter rdEventMeter = (RdEventMeter) rdEvent;
            taskEventData.setRdEventMeter(Lists.newArrayList(rdEventMeter));
            taskEventData.setDeviceId(rdEventMeter.getMeterId());
            taskEventData.setDeviceType(DeviceType.METER.getId());
            taskEventData.setIsODR(false);
        } else if (rdEvent instanceof RdEventDcu) {
            RdEventDcu rdEventDcu = (RdEventDcu) rdEvent;
            taskEventData.setRdEventDcu(Lists.newArrayList(rdEventDcu));
            taskEventData.setDeviceId(rdEventDcu.getDcuId());
            taskEventData.setDeviceType(DeviceType.DCU.getId());
            taskEventData.setIsODR(false);
        }
        kafkaTemplate.send(TaskKafkaTopic.DEVICE_EVENT_TRANSFER, ProtoBufferUtil.toProtoBuffer(taskEventData, TaskEventDataProto.TaskEventData.class).toByteArray());
        log.info("send event to DDS type:{} deviceId:{}", taskEventData.getDeviceType(), taskEventData.getDeviceId());
    }

    private List<RdEventMeterData> saveEventSubData(String seq, CosemData notificationDataValue, DeviceNotificationDto notificationItemList) {
        List<RdEventMeterData> rdEventMeterDataList = new ArrayList<>();
        for (int i = 0; i < notificationItemList.getObjectList().size(); i++) {
            GroupSubDataItem groupSubDataItem =notificationItemList.getObjectList().get(i);
            if (groupSubDataItem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.TIME_DATE.getCode())
                    || groupSubDataItem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.EVENT_CODE.getCode())) {
                continue;
            }
            Long subDataItemId = groupSubDataItem.getSubDataItemId();
            CosemData data = notificationDataValue.getChildAt(0).getChildAt(i);
            Object value = cosemXmlService.getData(data.getType(), data.getValue());

            RdEventMeterData rdEventMeterData = new RdEventMeterData();
            rdEventMeterData.setLogSeq(seq);
            rdEventMeterData.setSubDataItemId(subDataItemId);
            rdEventMeterData.setValue(String.valueOf(value));
            rdEventMeterDataList.add(rdEventMeterData);
//            rdEventMeterDataMapper.insert(rdEventMeterData);
        }
        return rdEventMeterDataList;
    }

    private List<RdEventDcuData> saveDcuEventSubData(String seq, Integer type, CosemData notificationDataValue, DeviceNotificationDto notificationItemList) {
        int index = 0;
        List<RdEventDcuData> rdEventDcuDataList = new ArrayList<>();
        for (GroupSubDataItem groupSubDataItem : notificationItemList.getObjectList()) {
            if (groupSubDataItem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.TIME_DATE.getCode())
                    || groupSubDataItem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.EVENT_CODE.getCode())) {
                index++;
                continue;
            }

            Long subDataItemId = groupSubDataItem.getSubDataItemId();
            CosemData data = notificationDataValue.getChildAt(0).getChildAt(index++);
            Object value = cosemXmlService.getData(data.getType(), data.getValue());
            if (needAsciiDecodeEventTypes.contains(type)) {
                value = new String(ByteUtils.hexStringToByte(Convert.toStr(value)), Charset.forName("utf-8"));
            }
            RdEventDcuData rdEventDcuData = new RdEventDcuData();
            rdEventDcuData.setLogSeq(seq);
            rdEventDcuData.setSubDataItemId(subDataItemId);
            rdEventDcuData.setValue(String.valueOf(value));
            rdEventDcuDataList.add(rdEventDcuData);
            // rdEventDcuDataMapper.insert(rdEventDcuData);
        }
        return rdEventDcuDataList;
    }

    private String saveRdAlarmMeter(Long meterId, Integer alarmObjId, Integer alarmBit, Date uploadDate, String taskNo) {
        RdAlarmMeter rdAlarmMeter = new RdAlarmMeter();
        String logSeq = UUID.randomUUID().toString();
        rdAlarmMeter.setLogSeq(logSeq);
        rdAlarmMeter.setMeterId(meterId);
        rdAlarmMeter.setAlarmObjId(alarmObjId);
        rdAlarmMeter.setAlarmBit(alarmBit);
        rdAlarmMeter.setTaskNo(taskNo);
        if(uploadDate != null) {
            rdAlarmMeter.setTvAlarm(uploadDate.getTime() / 1000);
        }
        int rtn = rdAlarmMeterMapper.insert(rdAlarmMeter);
        if (rtn >= 1) {
            return logSeq;
        }
        return null;
    }

    private Long getEventTv(Date uploadDate, CosemData notificationDataValue, DeviceNotificationDto notificationItemList, Long dataItemId) {
        int index = 0;
        if (notificationItemList == null) {
            return uploadDate.getTime() / 1000;
        }
        for (GroupSubDataItem groupSubDataItem : notificationItemList.getObjectList()) {
            if (groupSubDataItem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.TIME_DATE.getCode())) {
                CosemData data;
                if (dataItemId == null) {
                    data = notificationDataValue.getChildAt(0).getChildAt(index);
                } else {
                    data = notificationDataValue.getChildAt(index);
                }
                Object timeObj = cosemXmlService.getData(data.getType(), data.getValue());
                Date eventTv = null;
                if (timeObj instanceof String) {
                    eventTv = CosemUtils.parseOctStringDataTime((String) timeObj);
                } else if (timeObj instanceof Date) {
                    eventTv = (Date) timeObj;
                } else {
                    log.error("fail to transfer date from notification data");
                }
                if (eventTv != null) {
                    return eventTv.getTime() / 1000;
                }
            }
            index++;
        }
        return uploadDate.getTime() / 1000;
    }


    @Override
    public boolean batchInsertEvent(List<RdEventMeter> readsLps) {
        if (CollectionUtils.isEmpty(readsLps)) {
            return true;
        }
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("INSERT INTO RD_EVENT_METER (LOG_SEQ, METER_ID, EVENT_TYPE, LB_DATA_ITEM_ID, TV, TV_UPDATE, RECEIVE_TYPE) VALUES (?, ?, ?, ?, ?, ?, ?)");
        String sql = stringBuilder.toString();
        String eventDataSql = "INSERT INTO RD_EVENT_METER_DATA (LOG_SEQ, SUB_DATA_ITEM_ID, VALUE) VALUES (?, ?, ?)";

        try (Connection conn = dataSource.getConnection(); PreparedStatement eventPs = conn.prepareStatement(sql); PreparedStatement eventDataPs = conn.prepareStatement(eventDataSql)){
            conn.setAutoCommit(false);
            for (RdEventMeter batchAddMeterRead : readsLps) {
                String seq = UUID.randomUUID().toString();
                eventPs.setString(1, seq);
                eventPs.setLong(2, batchAddMeterRead.getMeterId());
                eventPs.setInt(3, batchAddMeterRead.getEventType());
                eventPs.setLong(4, batchAddMeterRead.getLbDataItemId());
                eventPs.setLong(5, batchAddMeterRead.getTv());
                eventPs.setLong(6, batchAddMeterRead.getTvUpdate());
                eventPs.setInt(7, batchAddMeterRead.getReceiveType());
                eventPs.addBatch();
                eventPs.executeBatch();
                if (CollUtil.isNotEmpty(batchAddMeterRead.getSubData())) {
                    for (RdEventMeterData subDatum : batchAddMeterRead.getSubData()) {
                        subDatum.setLogSeq(batchAddMeterRead.getLogSeq());
                        eventDataPs.setString(1, seq);
                        eventDataPs.setLong(2, subDatum.getSubDataItemId());
                        eventDataPs.setString(3, subDatum.getValue());
                        eventDataPs.addBatch();
                    }
                    eventDataPs.executeBatch();
                }
            }
            conn.commit();
            eventPs.close();
            eventDataPs.close();
            conn.close();
        } catch (SQLException throwables) {
            log.error("batch insert error", throwables);
            return false;
        }
        return true;

    }

    @Override
    public boolean batchInsertDcuEvent(List<RdEventDcu> readsLps) {
        if (CollectionUtils.isEmpty(readsLps)) {
            return true;
        }
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("INSERT INTO RD_EVENT_DCU (LOG_SEQ, DCU_ID, EVENT_TYPE, TV, TV_UPDATE, RECEIVE_TYPE) VALUES (?, ?, ?, ?, ?, ?)");
        String sql = stringBuilder.toString();
        String eventDataSql = "INSERT INTO RD_EVENT_DCU_DATA (LOG_SEQ, SUB_DATA_ITEM_ID, VALUE) VALUES (?, ?, ?)";
        try (Connection conn = dataSource.getConnection(); PreparedStatement eventPs = conn.prepareStatement(sql);  PreparedStatement eventDataPs = conn.prepareStatement(sql)){
            conn.setAutoCommit(false);
            for (RdEventDcu batchAddDcuRead : readsLps) {
                String seq = UUID.randomUUID().toString();
                eventPs.setString(1, seq);
                eventPs.setLong(2, batchAddDcuRead.getDcuId());
                eventPs.setLong(3, batchAddDcuRead.getEventType());
                eventPs.setLong(4, batchAddDcuRead.getTv());
                eventPs.setLong(5, batchAddDcuRead.getTvUpdate());
                eventPs.setInt(6, batchAddDcuRead.getReceiveType());
                eventPs.addBatch();
                if (CollUtil.isNotEmpty(batchAddDcuRead.getSubData())) {
                    for (RdEventDcuData subDatum : batchAddDcuRead.getSubData()) {
                        subDatum.setLogSeq(batchAddDcuRead.getLogSeq());
                        eventDataPs.setString(1, seq);
                        eventDataPs.setLong(2, subDatum.getSubDataItemId());
                        eventDataPs.setString(3, subDatum.getValue());
                        eventDataPs.addBatch();
                    }
                    eventDataPs.executeBatch();
                }
            }
            eventPs.executeBatch();
            conn.commit();
            eventPs.close();
            eventDataPs.close();
            conn.close();
        } catch (SQLException throwables) {
            log.error("batch insert error", throwables);
            return false;
        }
        return true;
    }
}
